package com.tang.lts.core.nio.processor;

import com.tang.lts.core.constant.Constants;
import com.tang.lts.core.factory.NameThreadFactory;
import com.tang.lts.core.nio.channel.ChannelInitializer;
import com.tang.lts.core.nio.channel.NioChannel;
import com.tang.lts.core.nio.handler.Futures;
import com.tang.lts.core.nio.handler.NioHandler;
import com.tang.lts.core.nio.idle.IdleDetector;
import com.tang.lts.core.nio.loop.NioSelectorLoop;
import com.tang.lts.core.support.SystemClock;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author Tang
 * @classname AbstractNioProcessor
 * @description [  ]
 * @date 2022/5/12 23:41
 */
public abstract class AbstractNioProcessor implements NioProcessor {

    private NioHandler eventHandler;
    private static AtomicBoolean started = new AtomicBoolean(false);

    protected ChannelInitializer channelInitializer;

    protected NioSelectorLoop selectorLoop;

    private Executor executors;

    protected IdleDetector idleDetector;

    private ConcurrentHashMap<NioChannel, WriteQueue> QUEUE_MAP = new ConcurrentHashMap<>();

    public AbstractNioProcessor(NioHandler eventHandler, ChannelInitializer channelInitialize) {
        this.eventHandler = eventHandler;
        this.channelInitializer = channelInitialize;
        // 创建 selectorLoop
        this.selectorLoop = new NioSelectorLoop("AcceptNioSelectorLoop-I/O",this);
        executors = Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR,
                new NameThreadFactory("NioServerProcessor", true));
        // 启动心跳检测
        this.idleDetector = new IdleDetector();
        this.idleDetector.start();
    }

    protected NioHandler eventHandler() {
        return this.eventHandler;
    }

    public void start() {
        if (started.compareAndSet(false, true)) {
            this.selectorLoop.start();
        }
    }

    @Override
    public void accept(SelectionKey key) {
        NioChannel channel = doAccept(selectorLoop);
        // 每一个socket专属的write queue
        QUEUE_MAP.put(channel, new WriteQueue());
    }

    /**
     * server 与 client 分别实现
     * @param selectorLoop
     * @return
     */
    public abstract NioChannel doAccept(NioSelectorLoop selectorLoop);

    @Override
    public Futures.WriteFuture writeFlush(NioChannel channel, Object msg) {
        SelectionKey selectionKey = channel.socketChannel().keyFor(selectorLoop.selector());
        if (selectionKey != null && selectionKey.isValid()) {
            selectionKey.interestOps(SelectionKey.OP_WRITE);
        }
        return write(channel,msg,true);
    }

    protected Futures.WriteFuture write(NioChannel channel, Object msg, boolean flush){
        Futures.WriteFuture future = Futures.newWriteFuture();
        if (msg == null) {
            future.setSuccess(true);
            future.notifyListeners();
        }
        // 编码
        ByteBuffer buffer = channel.getEncoder().encode(channel, msg);
        if (buffer == null) {
            future.setSuccess(false);
            future.notifyListeners();
        }
        // 放到队列里面
        QUEUE_MAP.get(channel).offer(new WriteRequest(buffer,future));
        if (flush) {
            doFlush(channel);
        }
        return future;
    }

    @Override
    public void read(NioChannel channel) {
        ByteBuffer readBuffer = ByteBuffer.allocate(64 * 1024);
        try {
            int readCount = channel.socketChannel().read(readBuffer);
            if (readCount < 0) {
                channel.close();
            } else if (readCount > 0) {

                readBuffer.flip();
                // 处理读
                doMessageReceive(channel, readBuffer);

                readBuffer.clear();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected void doMessageReceive(NioChannel channel, ByteBuffer readBuffer) {
        getExecutors().execute(() -> {
            try {
                List<Object> objs = channel.getDecoder().decoder(channel, readBuffer);
                if (objs != null) {
                    for (Object obj : objs) {
                        eventHandler().messageReceived(channel,obj);
                    }
                }
                channel.setLastReadTime(SystemClock.now());
            } catch (Exception e) {
                eventHandler().exceptionCaught(channel,e);
            }
        });
    }

    public Executor getExecutors() {
        return executors;
    }

    @Override
    public void flush(NioChannel channel) {
        doFlush(channel);
    }

    protected void doFlush(NioChannel channel){
        getExecutors().execute(() -> {
            WriteQueue writeQueue = QUEUE_MAP.get(channel);
            // 加锁 写出，，如果这个时候有其他写操作插入的话，头部数据不是会变吗，，这个步骤的加锁意义在那？
            if (!writeQueue.tryLock()) {
                return;
            }
            try {
                while (!writeQueue.isEmpty()) {
                    WriteRequest writeRequest = writeQueue.peek();
                    Futures.WriteFuture writeFuture = writeRequest.getWriteFuture();
                    try {
                        ByteBuffer buffer = writeRequest.getMessage();
                        try {
                            int writeCount = channel.socketChannel().write(buffer);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                        // 更新写时间
                        channel.setLastWriteTime(SystemClock.now());
                        if (buffer.remaining() == 0) {
                            // 弹出头部，会不会有问题，这个地方？
                            writeQueue.poll();
                            writeFuture.setSuccess(true);
                            writeFuture.notifyListeners();
                        } else {
                            break;
                        }
                    } catch (Exception e) {
                        writeFuture.setSuccess(false);
                        writeFuture.setCause(e);
                        writeFuture.notifyListeners();
                        eventHandler().exceptionCaught(channel,e);
                        break;
                    }
                }
                SelectionKey key = channel.socketChannel().keyFor(selectorLoop.selector());
                if (key != null && key.isValid()) {
                    key.interestOps(SelectionKey.OP_READ);
                }
            }finally {
                writeQueue.unlock();
            }
        });
    }

    @Override
    public Futures.ConnectFuture connect(SocketAddress remoteAddress) {
        Futures.ConnectFuture connectFuture = Futures.newConnectFuture();
        NioChannel channel = doConnect(remoteAddress,selectorLoop,connectFuture);
        connectFuture.setChannel(channel);
        QUEUE_MAP.put(channel, new WriteQueue());
        return connectFuture;
    }

    protected abstract NioChannel doConnect(SocketAddress remoteAddress, NioSelectorLoop selectorLoop, Futures.ConnectFuture connectFuture);
}